package com.tranq;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Hello world!
 *
 */
public class App 
{
    public static void main( String[] args )
    {
        try(Connection c = getConnection();
            Table table = createTable(c)){

            List<String> rowKeys = insertDatas(table);
            deleteDatas(table, rowKeys);

        }catch (Exception e){
            System.out.printf("some error: %s\n", e.getMessage());
        }
    }

    static Connection getConnection() throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "127.0.0.1");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.master", "127.0.0.1:60000");
        return ConnectionFactory.createConnection(configuration);
    }

    static Table createTable(Connection connection) throws IOException {
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create("tranq").build();

        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        // 为什么listNamespace 报错：listNamespace操作不支持？
        if(Arrays.stream(admin.listNamespaceDescriptors()).noneMatch(x -> Objects.equals(x.getName(), "tranq"))) {
            admin.createNamespace(namespaceDescriptor);
            System.out.println("create namespace [tranq]!");
        }else{
            System.out.println("namespace [tranq] has existed!");
        }
        TableName tableName = TableName.valueOf("tranq", "student");
        if(!admin.tableExists(tableName)){
            TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
                    .setColumnFamilies(Arrays.asList(
                            ColumnFamilyDescriptorBuilder
                                    .newBuilder("info".getBytes(StandardCharsets.UTF_8)).build(),
                            ColumnFamilyDescriptorBuilder
                                    .newBuilder("score".getBytes(StandardCharsets.UTF_8)).build()
                            ))
                    .build();
            admin.createTable(tableDescriptor);
            System.out.println("create new table [tranq:student]");
        }else{
            System.out.println("[tranq:student] has existed!");
        }
        return connection.getTable(tableName);
    }

    static List<String> insertDatas(Table table) throws Exception {

        try(TextDataFileLoader loader = new TextDataFileLoader("data.txt")){
            List<String> rowKeys = StreamSupport.stream(
                Spliterators.spliteratorUnknownSize(loader, Spliterator.ORDERED), false)
                    .map(insertOneRecord(table)).filter(Objects::nonNull)
                    .collect(Collectors.toList());
            System.out.println("insert datas successfully!");
            return rowKeys;
        }
    }

    static Function<String, String> insertOneRecord(Table table){
        byte[] infoFamily = "info".getBytes(StandardCharsets.UTF_8);
        byte[] scoreFamily = "score".getBytes(StandardCharsets.UTF_8);
        byte[] studentId = "student_id".getBytes(StandardCharsets.UTF_8);
        byte[] studentClass = "class".getBytes(StandardCharsets.UTF_8);
        byte[] understanding = "understanding".getBytes(StandardCharsets.UTF_8);
        byte[] programming = "programming".getBytes(StandardCharsets.UTF_8);

        return line -> {
            String[] fd = line.split(",");
            Put put = new Put(Bytes.getBytes(ByteBuffer.wrap(fd[0].getBytes(StandardCharsets.UTF_8))));
            switch (fd.length){
                case 5:
                    put.addColumn(scoreFamily, programming, fd[4].getBytes(StandardCharsets.UTF_8));
                case 4:
                    put.addColumn(scoreFamily, understanding, fd[3].getBytes(StandardCharsets.UTF_8));
                case 3:
                    put.addColumn(infoFamily, studentClass, fd[2].getBytes(StandardCharsets.UTF_8));
                case 2:
                    put.addColumn(infoFamily, studentId, fd[1].getBytes(StandardCharsets.UTF_8));
            }
            try {
                table.put(put);
            } catch (IOException e) {
                System.out.printf("insert [%s] error, %s%n", line, e.getMessage());
                return null;
            }
            return fd[0];
        };
    }

    static void deleteDatas(Table table, List<String> rowKeys){
        rowKeys.stream().forEach(key -> {
            Delete del = new Delete(key.getBytes(StandardCharsets.UTF_8));
            try {
                table.delete(del);
                System.out.printf("delete [%s] ok.\n", key);
            } catch (IOException e) {
                System.out.printf("delete [%s] error!\n", key);
            }
        });
    }
}
